package net.bwie.dt.job.Doris;

import net.bwie.dt.bean.OrderBean;
import net.bwie.dt.funtion.dtBeanMapFuctionOrder;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import utils.DorisUtil;
import utils.KafkaUtil;

public class OrderToDoris {
    public static void main(String[] args) throws Exception {
        // env执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);
        // 消费Kafka
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "dwd_n_order_pay_table");
//        kafkaStream.print();
        // 数据转换
        DataStream<String> resultStream = healde(kafkaStream);

        resultStream.print();
        // 存入Doris
        DorisUtil.saveToDoris(
                resultStream,
                "dt_realtime_ranking",
                "order_pay_table"
        );
        // 触发执行
        env.execute("OrderToDoris");
    }

    private static DataStream<String> healde(DataStream<String> stream) {
        // 转实体类
        SingleOutputStreamOperator<OrderBean> beanStream = stream.map(new dtBeanMapFuctionOrder());


        SingleOutputStreamOperator<String> process = beanStream.process(new ProcessFunction<OrderBean, String>() {
            @Override
            public void processElement(OrderBean value, Context ctx, Collector<String> out) throws Exception {
                int order_id = value.getOrder_id();
                int user_id = value.getUser_id();
                int product_id = value.getProduct_id();
                int quantity = value.getQuantity();
                Double price = value.getPrice();
                String status = value.getStatus();
                Double pay_amount = value.getPay_amount();
                String pay_time = value.getPay_time();
                int is_presale = value.getIs_presale();
                String presale_stage = value.getPresale_stage();
                String output = order_id + "," + user_id + "," + product_id + "," + quantity+ "," +
                        price+ "," + status+ "," + pay_amount+ "," + pay_time
                        + "," + is_presale+ "," + presale_stage;
                out.collect(output);
            }
        });
        return process;
    }
}
